#
# This file is a part of the open-eBackup project.
# This Source Code Form is subject to the terms of the Mozilla Public License, v. 2.0.
# If a copy of the MPL was not distributed with this file, You can obtain one at
# http://mozilla.org/MPL/2.0/.
#
# Copyright (c) [2024] Huawei Technologies Co.,Ltd.
#
# THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
# EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
# MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
#

import grp
import json
import os

import datetime
import pwd
import re
import shlex
import socket
import uuid

import psutil
import subprocess

from dws.commons.log_utils import log
from common.util.check_user_utils import check_os_user, check_path_owner
from common.util.cmd_utils import cmd_format
from common.util.exec_utils import exec_overwrite_file
from dws.commons.const import (AgentTypeEnum, DwsRetEnum, IntrusiveMode, DwsDeployType, PERMISSION_755, PERMISSION_444,
                               AUTH_METHOD_LIST,
                               CommonRegexp,
                               RoachConstant,
                               UserInfo)

from common.common import check_path_legal, execute_cmd, \
    check_command_injection_exclude_quote, execute_cmd_list, check_port_is_used
from common.env_common import get_install_head_path
from common.file_common import change_path_permission, exec_lchown_dir_recursively
from common.util.exec_utils import exec_mkdir_cmd, exec_umount_cmd, exec_mount_cmd, su_exec_rm_cmd

GDS_PROCESS = "gds"


def get_uid_gid(user_name):
    return pwd.getpwnam(user_name).pw_uid, pwd.getpwnam(user_name).pw_gid


def is_intrusive(nodes: list):
    """
    功能描述： 判断是不是非侵入式节点
    参数：@下发参数信息
    返回值：侵入式/非侵入式
    """
    for node in nodes:
        if node.get("extendInfo", {}).get("nodeType") == "1":
            return IntrusiveMode.NON_INTRUSIVE_MODE

    return IntrusiveMode.INTRUSIVE_MODE


def is_main_job(param: dict):
    if 'subJob' in param.keys():
        log.info('Sub job.')
        return False
    log.info('Main job.')
    return True


def get_cur_agent_id_from_param(nodes: list):
    """
    功能：从所有节点信息中获取当前主机的 agent id
    @nodes: 下发所有节点信息
    """
    cur_all_ip = set(get_cur_host_all_ip())
    for node in nodes:
        register_ip = set(node.get('extendInfo', {}).get("agentIpList", "").split(","))
        ip_info = cur_all_ip & register_ip
        if ip_info:
            agent_id = node.get("id", '')
            log.info(f'Get cur agent id: {agent_id}.')
            return agent_id
    log.error('Failed get cur agent id.')
    return ''


def is_proxy_node(nodes: list):
    """
    功能：判断当前节点是不是代理节点
    @nodes: 下发所有节点信息
    """
    cur_all_ip = get_cur_host_all_ip()
    for node in nodes:
        register_ip = node.get('endpoint', '')
        subnet_ip = node.get('extendInfo', {}).get('subNetFixedIp', '')
        if (register_ip in cur_all_ip or subnet_ip in cur_all_ip) and \
                int(node.get("extendInfo", {}).get("nodeType")) == AgentTypeEnum.PROXY:
            log.info('Proxy Node.')
            return True
    log.info('Cluster Node.')
    return False


def get_gds_process_param():
    """
    功能： 获取当前所有GDS进程的参数
    返回值： 包含GDS进程的参数列表
    """

    def get_single_param(process_pid):
        process = psutil.Process(process_pid)
        gds_cmds = process.cmdline()
        data_dir = ""
        ip_port = ""
        # 如果gds启动进程没有-t参数，设置默认值为1
        concurrent = 1
        user_name = process.username()
        for index, item in enumerate(gds_cmds):
            if "-d" in item:
                data_dir = gds_cmds[index + 1]
            if "-p" in item:
                ip_port = gds_cmds[index + 1]
            if "-t" in item:
                concurrent = int(gds_cmds[index + 1])
        if not ip_port:
            log.error(f"Process [{process_pid}] has no '-p' parameter.")
            raise Exception(f"Process [{process_pid}] has no '-p' parameter.")
        log.info(f"GDS process param [{user_name}], [{', '.join(gds_cmds)}].")
        return {"user": user_name, "ipPort": ip_port, "dataDir": data_dir, "concurrent": concurrent}

    ret_param_list = []
    pid_list = psutil.pids()
    for pid in pid_list:
        try:
            process_name = psutil.Process(pid).name()
        except Exception as err:
            log.warn(f"Get process err: {err}.")
            continue
        if GDS_PROCESS not in process_name:
            continue
        try:
            single_param = get_single_param(pid)
        except Exception as err:
            log.warn(f"Get single gds param err: {err}.")
            continue
        ret_param_list.append(single_param)

    if not ret_param_list:
        log.error("Get all GDS process param failed.")
        return False, []

    return True, ret_param_list


def check_mount_bind_point(mount_point):
    """
    检测挂载点是否有其他文件路径 mount bind
    :param mount_point: str
    :return: bool
    """
    check_cmd = ["mount", f"grep {mount_point}"]
    return_code, std_out, std_err = execute_cmd_list(check_cmd)
    if return_code != DwsRetEnum.SUCCESS:
        log.warning(f"Mount point have not found error {std_err}")
        return False
    log.debug("Mount point have found")
    return True


def umount_bind_path(mount_point):
    """
    去挂载指定挂载点
    :param mount_point: str
    :return: bool
    """

    return_code, std_out, std_err = exec_umount_cmd(mount_point)
    if return_code != DwsRetEnum.SUCCESS:
        log.warning(f"Failed to exec umount bind path, err: {std_err}")
        return False
    log.info("Succeed to exec umount bind path")
    return True


def umount_all_gds_data_dir():
    dws_config = get_dws_config()
    deploy_type = int(dws_config.get("deployType", DwsDeployType.DEPLOY_IN_SERVER.value))
    if deploy_type == DwsDeployType.DEPLOY_IN_SANDBOX.value:
        sandbox_path = dws_config.get("sandboxPath", "/var/chroot").strip()
        if not sandbox_path:
            sandbox_path = "/var/chroot"
        if check_command_injection_exclude_quote(sandbox_path):
            log.error(f"Sandbox path {sandbox_path} is invalid")
            return
        data_path = os.path.join(sandbox_path, "mnt/databackup")
        umount_bind_path(data_path)
    ret, gds_info = get_gds_process_param()
    if not ret:
        log.warning("Get gds info failed can not execute umount bind info")
        return

    for gds_item in gds_info:
        data_path = gds_item.get("dataDir", "")
        if not data_path:
            log.warning("GDS item hava no data dir info")
            continue
        umount_bind_path(data_path)


def mount_bind_path(src_area, des_area):
    """
    src_area 挂载绑定到指定的 des_area
    :param src_area: str
    :param des_area: str
    :return: bool
    """
    result = check_mount_bind_point(des_area)
    if result:
        log.info("Des path have mount other path")
        if not umount_bind_path(des_area):
            return False
    log.debug(f"Mount bind src path {src_area} to {des_area}")
    if not os.path.isdir(des_area):
        ret = exec_mkdir_cmd(des_area, mode=0x700)
        if not ret:
            log.error(f"Create dir {des_area} failed.")
            return False
    return_code, std_out, std_err = exec_mount_cmd(src_area, des_area)
    if return_code != DwsRetEnum.SUCCESS:
        log.error(f"Failed to exec mount bind path, err: {std_err}")
        return False
    return True


def del_path_without_exception(path: str):
    """
    函数功能：删除路径，不管是文件还是目录，并且不抛出异常
    参数：
    path: 文件路径
    返回值：
    成功返回True
    失败返回False
    """
    if not os.path.exists(path):
        log.warning(f"File [{path}] not exists.")
        return False

    path = os.path.realpath(path)
    if not su_exec_rm_cmd(path):
        log.warn(f"Fail to remove {path}.")
        return False

    log.info(f"Del path [{path}] suc.")
    return True


def check_sh_cmd_param(param):
    """
    检测拼接shell命令的参数是否含有特殊字符，防止命令注入
    :param  : 拼接 shell 的参数
    :return : bool
    """
    if not param or not isinstance(param, str):
        return False
    return not check_command_injection_exclude_quote(param)


def check_sql_cmd_param(param):
    """
    检测拼接sql命令的参数是否含有特殊字符，防止命令注入
    :param  : 拼接 sql 的参数
    :return : bool
    """
    if not param or not isinstance(param, str):
        return False
    # 特别注意，要匹配反斜杠(\)需要使用(\\\\)
    expression = "[|;&$><`!+%/\'\"\\\\:\[^]"
    if re.search(expression, param):
        return False
    return True


def check_sh_cmd_param_for_schema_and_table(param):
    """
    检测拼接shell命令的schema、table参数是否含有特殊字符，防止命令注入 (schema、table允许$符号)
    :param  : 拼接 shell 的参数
    :return : bool
    """
    if not param or not isinstance(param, str):
        return False
    expression = "[|;&><`'\"!+\n]"
    if re.search(expression, param):
        return False
    return True


def check_sql_cmd_param_for_schema_and_table(param):
    """
    检测拼接sql命令的schema、table参数是否含有特殊字符，防止命令注入 (schema、table允许$符号)
    :param  : 拼接 sql 的参数
    :return : bool
    """
    if not param or not isinstance(param, str):
        return False
    # 特别注意，要匹配反斜杠(\)需要使用(\\\\)
    expression = "[|;&><`!+%/\'\"\\\\:\[^]"
    if re.search(expression, param):
        return False
    return True


def check_path_valid(input_path, parent_dir="/mnt/databackup/"):
    """
    将路径转为绝对路径，并校验起是否在指定父目录下,防止目录逃逸
    :path : 需要检验的目录
    :parent_dir: 父目录
    :return: bool
    """
    if not input_path or not parent_dir:
        return False
    if not isinstance(input_path, str) or not isinstance(parent_dir, str):
        return False
    if not check_path_legal(input_path, parent_dir):
        log.error(f"Path {input_path} is invalid not start {parent_dir}")
        return False
    return not check_command_injection_exclude_quote(input_path)


def get_result_path(user_name: str):
    # 获取结果文件的路径
    if not check_os_user(user_name):
        log.error(f"Invalid user: {user_name}")
        return ""
    get_pwd_path = cmd_format("su - {} -c 'pwd'", user_name)
    ret, output, error = execute_cmd(get_pwd_path)
    output = output.split("\n")
    if ret and output:
        result_path = os.path.join(output[0].strip(), f'result-{uuid.uuid4()}.txt')
    else:
        result_path = os.path.join("/home", user_name, f'result-{uuid.uuid4()}.txt')
    if os.path.exists(result_path):
        log.error(f"Result path [{result_path}] already exist.")
        delete_result_path(result_path)

    return result_path


def read_result_file(result_path: str):
    abs_path = os.path.realpath(result_path)
    result = ""
    if not os.path.isfile(abs_path):
        log.error(f"Result file [{abs_path}] not exist.")
        return False, result
    if not check_path_valid(abs_path, "/home/"):
        log.error(f"Result path [{abs_path}] invalid.")
        return False, result
    try:
        with open(abs_path, "r", encoding='utf-8') as tmp:
            result = tmp.read()
    except Exception:
        log.error(f"Read result file [{abs_path}] err.")
        return False, ""
    return True, result


def delete_result_path(result_path: str):
    abs_path = os.path.realpath(result_path)
    if os.path.isfile(abs_path) and check_path_valid(abs_path, "/home/"):
        if not su_exec_rm_cmd(abs_path):
            log.error(f"Fail to remove {abs_path}.")


def get_dws_config():
    dws_conf = f"{get_install_head_path()}" \
               f"/DataBackup/ProtectClient/Plugins/GeneralDBPlugin/bin/applications/dws/dws.conf"
    try:
        with open(dws_conf, "r") as temp_f:
            dws_config = json.loads(temp_f.read())
    except Exception as err:
        log.error("Fail to read dws conf for %s", err)
        return ""
    return dws_config


def handle_special_cmd(cmd: str):
    special_cmd_list = ["$GPHOME", "$GAUSS"]
    new_cmd = cmd
    is_special = False
    separator = '\\\\'
    for single in special_cmd_list:
        if single in new_cmd:
            is_special = True
            pos = new_cmd.find(single)
            new_cmd = new_cmd[:pos] + separator + new_cmd[pos:]
    if not is_special:
        new_cmd = new_cmd.replace("\\", "\\\\")
        new_cmd = new_cmd.replace("\"", "\\\"")
    return new_cmd


def construct_dws_cmd(user: str, cmd: str, env_file: str):
    """
    功能说明：通过判断不同的部署形态组装不同的执行命令
    user： dws集群用户
    cmd：要执行的命令
    env_file：要加载的环境变量
    """
    dws_config = get_dws_config()
    deploy_type = int(dws_config.get("deployType", DwsDeployType.DEPLOY_IN_SERVER.value))
    if not check_os_user(user):
        log.error(f"Invalid user: {user}")
        return ""
    if deploy_type == DwsDeployType.DEPLOY_IN_SERVER.value:
        result = f"su - {user} -c \"source {env_file} && {cmd}\""
    elif deploy_type == DwsDeployType.DEPLOY_IN_SANDBOX.value:
        sandbox_ip = dws_config.get("loginSandboxIp", "").strip()
        if not sandbox_ip:
            log.error("Fail to get sandbox ip")
            return ""
        cmd = handle_special_cmd(cmd)
        result = f"su - {user} -c \"ssh {sandbox_ip} -C \\\"source {env_file};source /etc/profile;{cmd}\\\"\""
    else:
        log.error("Unknown deploy type %s", str(deploy_type))
        return ""
    log.debug("Construct dws cmd: %s", result)
    return result


def construct_dws_array_cmd(user: str, cmd: str, env_file: str):
    """
    功能说明：通过判断不同的部署形态组装不同的数组命令，当cmd中有$符号需调用此方法构建数组命令，因为$符号会导致普通命令执行失败
    :param user: dws集群用户
    :param cmd: 要执行的命令
    :param env_file:要加载的环境变量
    :return: 数组命令
    """
    dws_config = get_dws_config()
    deploy_type = int(dws_config.get("deployType", DwsDeployType.DEPLOY_IN_SERVER.value))
    if not check_os_user(user):
        log.error(f"Invalid user: {user}")
        return []
    if deploy_type == DwsDeployType.DEPLOY_IN_SERVER.value:
        result = ['su', '-', f'{user}', '-c', f'source {env_file} && {cmd}']
    elif deploy_type == DwsDeployType.DEPLOY_IN_SANDBOX.value:
        sandbox_ip = dws_config.get("loginSandboxIp", "").strip()
        if not sandbox_ip:
            log.error("Fail to get sandbox ip")
            return []
        cmd = cmd.replace("$", "\\$").replace("\"", "\\\"")
        result = ['su', '-', f'{user}', '-c', f'ssh {sandbox_ip} -C "source {env_file};source /etc/profile;{cmd}"']
    else:
        log.error("Unknown deploy type %s", str(deploy_type))
        return []
    log.debug("Construct dws array cmd: %s", result)
    return result


def dws_exec_cmd_by_type(user: str, env_file: str, cmd: str):
    # 组装命令
    exec_cmd = construct_dws_cmd(user, cmd, env_file)
    if not exec_cmd:
        log.error("Construct dws cmd failed")
        return False, "Construct dws cmd failed"
    result_path = get_result_path(user)
    ret, output = dws_exec_cmd(exec_cmd, result_path)
    return ret, output


def dws_exec_array_cmd_by_type(user: str, env_file: str, cmd: str):
    # 组装命令
    exec_cmd = construct_dws_array_cmd(user, cmd, env_file)
    if not exec_cmd:
        log.error("Construct dws array cmd failed")
        return False, "Construct dws array cmd failed"
    result_path = get_result_path(user)
    ret, output = dws_exec_cmd(exec_cmd, result_path)
    return ret, output


def dws_exec_sql_cmd_by_type(user: str, env_file: str, sh_cmd: str, sql_cmd: str):
    # 组装命令
    exec_cmd = construct_dws_cmd(user, sh_cmd, env_file)
    if not exec_cmd:
        log.error("Construct dws cmd failed")
        return False, "Construct dws cmd failed"
    result_path = get_result_path(user)
    ret, output = dws_exec_cmd(exec_cmd, result_path, sql_cmd)
    return ret, output


def exec_cmd_by_cmd_type(sh_cmd, sql_cmd: str, result_path: str):
    """
    执行普通命令或数组命令
    """
    if isinstance(sh_cmd, str):
        if sql_cmd:
            cmd_ret, _, _ = execute_sql_cmd(sh_cmd, sql_cmd, result_path)
        else:
            sh_cmd = sh_cmd[:-1] + f" >& {result_path}" + sh_cmd[-1]
            cmd_ret, _, _ = execute_cmd(sh_cmd)
    elif isinstance(sh_cmd, list):
        sh_cmd[-1] += f" >& {result_path}"
        cmd_ret, _, _ = execute_cmd(sh_cmd, cmd_array_flag=True)
    else:
        log.error("sh_cmd type is invalid")
        cmd_ret = DwsRetEnum.FAILED
    return cmd_ret


def execute_sql_cmd(sh_cmd: str, sql_cmd: str, result_path: str):
    """
    执行带有sql命令的shell命令，以标准输入方式输入sql命令防止sql中$符号导致shell命令执行失败
    """
    with open(result_path, 'w') as result_file:
        sh_cmd = shlex.split(sh_cmd)
        process = subprocess.Popen(sh_cmd, stdin=subprocess.PIPE, stdout=result_file, stderr=result_file,
                                   encoding=None, errors='ignore')
        process.stdin.write(sql_cmd)
        process.stdin.close()
        process.wait()
        ret_code = process.returncode
        return str(ret_code), "", ""


def dws_exec_cmd(sh_cmd, result_path: str, sql_cmd: str = None):
    """
    执行指定命令，并将结果重定向到文件，读取结果文件并过滤指定字符，最后删除结果文件，只有切用户执行的命令才能重定向到文件
    参数：
    sh_cmd： 待执行的命令，可为普通命令或数组命令
    result_path：结果文件的路径
    返回值：
    bool, list
    """
    out_info = ""
    tmp_sh_cmd = " ".join(sh_cmd) if isinstance(sh_cmd, list) else sh_cmd
    if not tmp_sh_cmd.startswith("su - ") or not result_path:
        log.error("Invalid param.")
        return False, out_info
    try:
        cmd_ret = exec_cmd_by_cmd_type(sh_cmd, sql_cmd, result_path)
    except Exception:
        log.error("Execute shell cmd exception.")
        delete_result_path(result_path)
        return False, out_info
    read_ret, out_info = read_result_file(result_path)
    delete_result_path(result_path)
    if not read_ret:
        log.error("Read result file failed.")
        return False, out_info
    if cmd_ret != DwsRetEnum.SUCCESS:
        log.error("Execute shell cmd failed.")
        return False, out_info
    dws_config = get_dws_config()
    deploy_type = int(dws_config.get("deployType", DwsDeployType.DEPLOY_IN_SERVER.value))
    if deploy_type == DwsDeployType.DEPLOY_IN_SANDBOX.value:
        out_info = out_info.split("\n")
        auth_next_index = 0
        line_num = 0
        for each_line in out_info:
            if "Authorized users only" in each_line:
                auth_next_index = line_num + 1
                break
            line_num += 1
        out_info = out_info[auth_next_index:]
        out_info = "\n".join(out_info)
    log.info("DWS exec shell cmd suc.")
    return True, out_info


def get_cur_host_all_ip() -> list:
    """
    获取当前机器上所有ipv4的地址，除了localhost
    """
    ip_list = []
    for net_card_property in psutil.net_if_addrs().values():
        for item in net_card_property:
            if len(item) < 2:
                continue
            if item[0] == socket.AF_INET and item[1] and not item[1] == '127.0.0.1':
                ip_list.append(item[1])

    return ip_list


def get_roach_ip_and_port(process_cmdline):
    """
    功能： 获取当前roach_client 运行的端口和ip
    参数： process_cmdline; roach 进程的运行参数
    返回值： 端口和ip
    """
    if not process_cmdline:
        return '', ''

    roach_ip, port = '', ''
    for index, item in enumerate(process_cmdline):
        if item == '-p':
            roach_ip = process_cmdline[index + 1].split(':')[0]
            port = process_cmdline[index + 1].split(':')[1]
            return roach_ip, port

    return roach_ip, port


def get_process_cmd(process_name):
    """
    :param process_name: 进程名称
    :return: 返回进程运行命令，包含参数等
    """
    cmdline = ''
    for pid in psutil.pids():
        try:
            tmp_name = psutil.Process(pid).name()
        except Exception as err:
            log.warn(f"Get process name err: {err}")
            continue
        try:
            pid_username = psutil.Process(pid).username()
        except Exception as err:
            log.warn(f"Get process username err: {err}")
            continue
        if process_name != tmp_name or (tmp_name == "roach_client" and pid_username != "rdadmin"):
            continue
        try:
            cmdline = psutil.Process(pid).cmdline()
        except Exception as err:
            log.warn(f"Get command line err:{err}")
            continue

    log.info(f"Ret: {cmdline}")
    return cmdline


def is_gds_same_user(param_list, name_key="user"):
    """
    检查所有GDS进程用户是否是同一用户启动
    """
    use_set = set()
    for item in param_list:
        use_set.add(item.get(name_key))

    if len(use_set) != 1:
        log.error("Not all gds user are the same.")
        return False
    return True


def is_gds_bind_unique_dir(param_list, dir_key="localDataDir"):
    """
    检查所有GDS进程是否绑定了不同的数据目录
    """
    data_dir = set()
    for item in param_list:
        data_dir.add(item.get(dir_key))

    if len(data_dir) != len(param_list):
        log.error("Some gds process data dir are the same.")
        return False
    return True


def empty_dir(path):
    """
    清空目录
    """
    if not check_path_valid(path):
        log.error(f"Invalid path [{path}].")
        return False
    del_success = True
    for temp_dir in os.listdir(path):
        del_path = os.path.join(path, temp_dir)
        del_path = os.path.realpath(del_path)
        if not su_exec_rm_cmd(del_path):
            log.error(f"Fail to remove {del_path}.")
            del_success = False
    return del_success


def record_subtask_info_to_file(host_ip: str, subtask_name: str, dir_path: str):
    """
    将子任务执行结果通过文件记录下来，方便在其他子任务中读取。
    """
    if not check_path_valid(dir_path):
        log.error(f"Invalid dir path [{dir_path}].")
        return False
    file_path = os.path.join(dir_path, f"{subtask_name}-{host_ip}.task")
    if os.path.isdir(file_path):
        log.error(f"File path [{file_path}] is a dir.")
        return False
    if os.path.isfile(file_path):
        log.error(f"File path [{file_path}] already exists.")
        return True

    try:
        os.mknod(file_path)
    except Exception as err:
        log.exception(f"Create file err: {err}.")
        return False

    log.info("Record subtask info to file success.")
    return True


def get_subtask_info_from_file(subtask_name: str, dir_path: str):
    """
    遍历目录下所有文件名，找到包含指定子任务名的文件名，拆分出ip，记录到list中。
    返回值：list
    """
    ret_list = []

    if not check_path_valid(dir_path):
        log.error(f"Invalid dir path [{dir_path}].")
        return ret_list
    for file_name in os.listdir(dir_path):
        if subtask_name in file_name:
            temp_dir = os.path.splitext(file_name)[0].split("-")
            if len(temp_dir) < 2:
                continue
            ret_list.append(temp_dir[1])

    log.info("Get subtask info from file success.")
    return ret_list


def check_subtask_info(cluster_nodes: list, agent_nodes: list, passed_nodes: list):
    """
    检查上个子任务的执行情况，用来指导当前任务是否能继续。
    """
    log.info(f"Cluster nodes: {cluster_nodes}, agent nodes: {agent_nodes}, passed nodes: {passed_nodes}.")
    if not agent_nodes:  # 侵入式
        return set(cluster_nodes) == set(passed_nodes)

    # 非侵入式
    if set(cluster_nodes) & set(passed_nodes) and set(agent_nodes) & set(passed_nodes):
        return True

    return False


def copy_file_into_sandbox(user: str, src_path: str, target_path: str):
    dws_config = get_dws_config()
    deploy_type = int(dws_config.get("deployType", DwsDeployType.DEPLOY_IN_SERVER.value))
    sandbox_ip = dws_config.get("loginSandboxIp", "").strip()
    if not check_path_owner(src_path, [user]):
        log.error(f"Check path owner failed: {src_path}")
        return False
    if deploy_type == DwsDeployType.DEPLOY_IN_SANDBOX.value:
        copy_cmd = cmd_format("su - {} -c \"scp {} {}:{}\"", user, src_path, sandbox_ip, target_path)
        ret, _, err = execute_cmd(copy_cmd)
        if ret != DwsRetEnum.SUCCESS:
            log.error(f"Fail to copy {src_path} to {target_path} for {err}")
            return False
    return True


def open_non_secret(user: str, env_file: str):
    # 开启免密
    dws_config = get_dws_config()
    auth_method = dws_config.get("authmethod", "").strip()
    if auth_method not in AUTH_METHOD_LIST:
        log.warning("Auth method is not configured.")
        return True
    if auth_method == "trust":
        log.info("Auth method is trust.")
        return True
    non_secret_cmd = "gs_guc set -Z coordinator -Z datanode -N all -I all -h \\\"local all all trust\\\""
    ret, output = dws_exec_cmd_by_type(user, env_file, non_secret_cmd)
    if not ret:
        log.error(f"Open non secret failed for {output}")
        return False
    return True


def close_non_secret(user: str, env_file: str):
    # 关闭免密
    dws_config = get_dws_config()
    auth_method = dws_config.get("authmethod", "").strip()
    if auth_method not in AUTH_METHOD_LIST:
        log.warning("Auth method is not configured.")
        return True
    if auth_method == "trust":
        log.info("Auth method is trust.")
        return True
    non_secret_cmd = f"gs_guc set -Z coordinator -Z datanode -N all -I all -h \\\"local all all {auth_method}\\\""
    ret, output = dws_exec_cmd_by_type(user, env_file, non_secret_cmd)
    if not ret:
        log.error(f"Close non secret failed for {output}")
        return False
    return True


def check_mount_point(mount_point: str):
    partitions = psutil.disk_partitions()
    for partition in partitions:
        if partition.mountpoint == mount_point:
            log.info(f"Mount point {mount_point} has mounted")
            return True
    return False


def select_available_port(start_port, end_port):
    for tmp_port in range(start_port, end_port):
        if not check_port_is_used(tmp_port):
            return tmp_port
        log.debug(f"Port {tmp_port} is used.")
    return 0


def mount_bind_sandbox_path():
    dws_config = get_dws_config()
    sandbox_path = dws_config.get("sandboxPath", "/var/chroot").strip()
    if not sandbox_path:
        sandbox_path = "/var/chroot"
    if check_command_injection_exclude_quote(sandbox_path):
        log.error(f"Sandbox path {sandbox_path} is invalid")
        return False
    if not os.path.exists("/mnt/databackup"):
        ret = exec_mkdir_cmd("/mnt/databackup")
        if not ret:
            log.error(f"Create mount path /mnt/databackup failed.")
            return False
        log.info("Create /mnt/databackup path success.")
    change_path_permission("/mnt", mode=PERMISSION_755)
    change_path_permission("/mnt/databackup", mode=PERMISSION_755)

    data_path = os.path.join(sandbox_path, "mnt/databackup")

    if not os.path.exists(data_path):
        ret = exec_mkdir_cmd(data_path, is_check_white_list=False)
        if not ret:
            log.error(f"Create path {data_path} is failed.")
            return False
        log.info("Mkdir mnt path suc.")
    elif check_mount_point(data_path):
        log.warn(f"Mount Point {data_path} has been mount")
        return True

    mnt_path = os.path.join(sandbox_path, 'mnt')
    change_path_permission(mnt_path, user_name="root", mode=PERMISSION_755)

    change_path_permission(data_path, user_name="root", mode=PERMISSION_755)
    return_code, _, std_err = exec_mount_cmd("/mnt/databackup", data_path)
    if return_code != DwsRetEnum.SUCCESS:
        log.error(f"Failed to exec mount bind path, err: {std_err}")
        return False
    return True


def is_in_interval(number, start, end):
    if start <= number <= end:
        return True
    return False


def is_roach_param_valid(param_key, param_value):
    compare_ret = True
    if param_key == "--cpu-cores":
        compare_ret = param_value.isdigit() and is_in_interval(int(param_value), 1, 1024)
    if param_key == "--master-timeout":
        compare_ret = param_value.isdigit() and is_in_interval(int(param_value), 600, 3600)
    if param_key == "--logging-level":
        compare_ret = param_value in ["fatal", "error", "warning", "info", "debug", "debug2"]
    if param_key == "--max-backup-io-speed":
        compare_ret = param_value.isdigit() and is_in_interval(int(param_value), 1, 2048)
    if param_key == "--parallel-process":
        compare_ret = param_value.isdigit() and is_in_interval(int(param_value), 1, 32)
    return compare_ret and is_roach_param_valid_two(param_key, param_value)


def is_roach_param_valid_two(param_key, param_value):
    compare_ret = True
    if param_key == "--filesplit-size":
        compare_ret = param_value.isdigit() and is_in_interval(int(param_value), 0, 1024)
    if param_key == "--reader-thread-count":
        compare_ret = param_value.isdigit() and is_in_interval(int(param_value), 1, 128)
    if param_key == "--reader-thread-file-count":
        compare_ret = param_value.isdigit() and is_in_interval(int(param_value), 1, 65535)
    if param_key == "--reader-thread-file-size":
        compare_ret = param_value.isdigit() and is_in_interval(int(param_value), 1, 65535)
    if param_key == "--log-filecount":
        compare_ret = param_value.isdigit() and is_in_interval(int(param_value), 0, 1024)
    if param_key == "--compression-level":
        compare_ret = param_value.isdigit() and is_in_interval(int(param_value), 0, 9)
    if param_key == "--compression-type":
        compare_ret = param_value.isdigit() and is_in_interval(int(param_value), 0, 2)
    return compare_ret


def construct_roach_param():
    """
    从dws.conf中获取roach参数配置
    """
    default_ret = " --filesplit-size 256 --parallel-process 16 "
    params_list = [
        "--cpu-cores", "--master-timeout", "--logging-level", "--max-backup-io-speed", "--parallel-process",
        "--filesplit-size", "--reader-thread-count", "--reader-thread-file-count",
        "--reader-thread-file-size", "--log-filecount", "--compression-type", "--compression-level"
    ]
    dws_config = get_dws_config()
    roach_param = dws_config.get("roachParam", {})
    if not roach_param:
        log.warning("No roach param config.")
        return default_ret
    param_str = ""
    for key, value in roach_param.items():
        if key not in params_list:
            log.warning(f"Invalid roach param key {key}")
            continue
        value = value.strip()
        if not value:
            continue
        if not is_roach_param_valid(key, value):
            log.warning(f"Param {key} is invalid.")
            continue
        param_str += f" {key} {value.strip()} "
    return param_str


def check_param_with_regexp(str_param, regexp):
    """
    使用正则校验字符串
    @param str_param: 校验入参字符串
    @param regexp: 正则表达式
    @return: 是否有效字符串
    """
    if not str_param or not isinstance(str_param, str):
        return False
    return bool(re.search(regexp, str_param))


def valid_backup_id(str_param):
    """
    校验backup_id是否合法
    @param str_param: 字符串参数
    @return: uuid是否合法
    """
    return check_param_with_regexp(str_param, CommonRegexp.backup_id_reg)


def has_uppercase(string):
    return any(char.isupper() for char in string)


def has_special_character(string):
    return any(char == '$' or char == '#' for char in string)


def get_file_attribute(file_name, job_id):
    """
    获取指定文件/目录归属的用户，组
    """
    object_file_attribute = os.stat(file_name)
    log.info(f"Get user{object_file_attribute.st_uid} group({object_file_attribute.st_gid}) success. "
             f"main task:{job_id}")
    return object_file_attribute.st_uid, object_file_attribute.st_gid


def set_file_attribute(path, user_id, group_id, job_id):
    """
    设置指定文件/目录归属的用户，组
    """
    user_name = pwd.getpwuid(user_id).pw_name
    group_name = grp.getgrgid(group_id).gr_name
    if not exec_lchown_dir_recursively(path, user_name, group_name):
        log.error(f"Fail to change user/group for {path}. main task:{job_id}")
        return False
    return True


def check_command_injection(param):
    """
    检测是否含有特殊字符防止命令注入
    :param : shell 执行参数
    :return: bool
    """

    expression = "[|;&$><`!+\n]"
    if re.search(expression, param):
        return True
    return False


def check_path_in_white_list(path_: str):
    try:
        real_path = os.path.realpath(path_)
    except Exception as e:
        log.error(f"Path verification error: {e}.")
        return False, ''
    if check_command_injection(real_path):
        log.error("Invalid path.")
        return False, ''
    for path in RoachConstant.WHITE_FILE_LIST:
        if real_path.find(path) == 0:
            return True, real_path
    log.error("The path is not in the trustlist.")
    return False, ''


def get_uid_and_gid(user_name, group_name):
    return pwd.getpwnam(user_name).pw_uid, pwd.getpwnam(group_name).pw_gid


def set_user_and_group(path, user, group):
    ret, real_path = check_path_in_white_list(path)
    if not ret:
        return False
    try:
        uid, gid = get_uid_and_gid(user, group)
    except Exception as e_info:
        log.error(f"Get path user and group error: {e_info}.")
        return False
    os.lchown(real_path, int(uid), int(gid))
    return True


def set_permisson(path, permisson):
    ret, real_path = check_path_in_white_list(path)
    if not ret:
        return False
    os.chmod(real_path, permisson)
    return True


def save_business_config(business_type, cache_path, copy_id):
    """
        功能描述：任务类型写入cach仓tmp/business_config.txt
        参数：
        返回值: True or False
    """
    log.info(f"business_type: {business_type}, cache_path: {cache_path}, copy_id: {copy_id}")
    business_type_info_dir = os.path.join(cache_path, "tmp", copy_id)
    if not os.path.exists(business_type_info_dir):
        if not exec_mkdir_cmd(business_type_info_dir):
            return False
    business_type_info = {"jobType": business_type}
    business_type_info_path = os.path.join(cache_path, "tmp",
                                           copy_id, RoachConstant.BUSINESS_CONFIG_FILE)
    log.info(f"save bussiness config {business_type_info} to {business_type_info_path}")
    exec_overwrite_file(business_type_info_path, business_type_info)
    if not os.path.exists(business_type_info_path):
        log.error("Create business_config.txt failed.")
        return False
    set_user_and_group(business_type_info_path, UserInfo.USER_RDADMIN, UserInfo.USER_RDADMIN)
    # 非侵入式部署场景下business_config.txt在dws节点上产生，在外置代理上被xbsaServer访问
    # 2个节点rdadmin用户的id不一样，需要赋予444权限
    set_permisson(business_type_info_path, PERMISSION_444)
    log.info("Write business_config.txt succ")
    return True


def process_table_name(table_name):
    schema_table_pair = table_name.split(".")
    if len(schema_table_pair) != 2:
        log.error(f"Invalid table name: [{table_name}].")
        return ""
    schema_name = schema_table_pair[0]
    table_name = schema_table_pair[1]
    return r'"' + schema_name + r'"."' + table_name + r'"'


def process_table_name_in_table_file(table_name):
    schema_table_pair = table_name.split(".")
    if len(schema_table_pair) != 2:
        log.error(f"Invalid table name: [{table_name}].")
        return ""
    schema_name = schema_table_pair[0]
    table_name = schema_table_pair[1]
    schema_name = r'"' + schema_name + r'"' if has_special_character(schema_name) or has_uppercase(schema_name) \
        else schema_name
    table_name = r'"' + table_name + r'"' if has_special_character(table_name) or has_uppercase(table_name) \
        else table_name
    return schema_name + r'.' + table_name


def convert_to_timestamp(datetime_str):
    dt = datetime.datetime.strptime(datetime_str, '%Y-%m-%d %H:%M:%S')
    timestamp = dt.timestamp()
    return int(timestamp)


def split_line_in_table_list_file(line):
    """
    拆分schema.list或table.list中的一行内容，schema\table可能包含双引号，根据双引号的"."拆分
    :param : 一行内容
    :return: list
    """
    # "[^"]*"匹配双引号内的所有字符，包括双引号本身, \w+表示匹配一个或多个单词字符，包括"_"
    pattern = r'"[^"]*"|\w+'
    result = re.findall(pattern, line)
    return result


def compare_version(version1, version2):
    """
    比较版本号。如果version1 < version2, 返回-1；如果version1 > version2, 返回1；如果相等，返回0。
    """
    v1_parts = split_version(version1)
    v2_parts = split_version(version2)
    v1_tuple = tuple(v1_parts)
    v2_tuple = tuple(v2_parts)
    if len(v1_tuple) != 3 or len(v2_tuple) != 3:
        raise Exception("Invalid version")
    if v1_tuple > v2_tuple:
        return 1
    elif v1_tuple < v2_tuple:
        return -1
    else:
        return 0


def split_version(version):
    parts = version.split('.')
    return [int(part) for part in parts]
